package com.deep.flink

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.api.Expressions.$

/**
 *
 * @param user
 * @param url
 * @param timeLength
 */
case class Event2(user: String, url: String, timeLength: Long)


object SimpleTableExample {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 读取数据源，创建DataStream
    val eventStream = env.fromElements(
      Event2("Alice", "./home", 1000L),
      Event2("Bob", "./cart", 1000L),
      Event2("Alice", "./prod?id=1", 5 * 1000L),
      Event2("Cary", "./home", 60 * 1000L),
      Event2("Bob", "./prod?id=3", 90 * 1000L),
      Event2("Alice", "./prod?id=7", 105 * 1000L)
    )

    // 创建表环境
    val tableEnv = StreamTableEnvironment.create(env)

    // 将DataStream转换成表
    val eventTable: Table = tableEnv.fromDataStream(eventStream)

    // 调用Table API进行转换计算
    /**
     * 这里的$符号是 Table API 中定义的“表达式”类 Expressions 中的一个静态方法，传入一
     * 个字段名称，就可以指代数据中对应字段，这个方法需要使用如下的方式进行手动导入。
     * import org.apache.flink.table.api.Expressions.$
     */
    val resultTable = eventTable.select($("user"), $("url"))
      .where($("user").isEqual("Alice"))

    // 直接写SQL SQL后直接 + eventTable，会自动注册一张和变量名同名的表
    val resultSqlTable = tableEnv.sqlQuery("select user, url from " + eventTable + " where user = 'Bob'")

    // 转换成流打印输出
    resultTable.printSchema()

    /**
     * resultStream> +I[Alice, ./home]
     * resultSqlStream> +I[Bob, ./cart]
     * resultStream> +I[Alice, ./prod?id=1]
     * resultSqlStream> +I[Bob, ./prod?id=3]
     * resultStream> +I[Alice, ./prod?id=7]
     * +I:这是表示每条数据都是“插入”（Insert）到表中的新增数据。
     */
    val resultStream = tableEnv.toDataStream(resultTable)
    resultStream.print("resultStream")

    val resultSqlStream = tableEnv.toDataStream(resultSqlTable)
    resultSqlStream.print("resultSqlStream")

    env.execute("simple table example")
  }

}